package com.xiaochong.tir.apiservice.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.service.common.config.redis.RedisServer;
import com.xiaochong.tir.apiservice.component.HbaseBaseQueryComponent;
import com.xiaochong.tir.apiservice.component.hbase.CurrencyInfoHbaseComponent;
import com.xiaochong.tir.common.data.hbasepo.TradesToKLine;
import com.xiaochong.tir.common.data.mapper.CurrencyInfoHbaseMapper;
import com.xiaochong.tir.common.exception.hbase.HbaseCloseExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseConnectionExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseGetTableExcetion;
import com.xiaochong.tir.common.util.DateUtils;
import com.xiaochong.tir.common.util.HbaseTableUtils;
import com.xiaochong.tir.common.util.KLineTypeEnum;
import com.xiaochong.tir.common.util.RedisKeysUtils;
import net.sf.json.util.JSONTokener;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Time;
import java.time.DayOfWeek;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Map;

import com.xiaochong.tir.common.data.po.KLine;
import com.xiaochong.tir.common.data.mapper.KLineMapper;

/**
 * 此类为[基础代码]自动生成,已经拥有基本的增删改成操作 
 * <其他请自行扩展>
 * @author cx
 * @time 2018-06-14 08:54
 * 
 */
@Service
public class KLineService{
 
      
    @Autowired
	private KLineMapper kLineDao;

    @Autowired
    private RedisServer redisServer;

    @Autowired
    private HbaseBaseQueryComponent hbaseBaseQueryComponent;

    @Autowired
    private CurrencyInfoHbaseMapper currencyInfoHbaseMapper;

    @Autowired
    private CurrencyInfoHbaseComponent currencyInfoHbaseComponent;
	/**
     * 新增
     * @param entity 
     */
    public int insertSelective (KLine entity) { 
     return this.kLineDao.insertSelective(entity);
    }	
    
    /**
     * 批量新增
     * @param list 
     */
    public void insertBatch(List<KLine> list) {
           
          this.kLineDao.insertBatch(list);
         
         
    }
    
    /**
     * 根据主键修改单条数据
     * @param entity 
     */
    public int updateByPrimaryKey(KLine entity) {
           return this.kLineDao.updateByPrimaryKey(entity);
    }
    
    /**
     * 根据主键批量修改数据
     * @param list 
     */
    public void updateBatch(List<KLine> list) {
           
          this.kLineDao.updateBatch(list);
          
        
    }
    
    /**
     * 根据主键查询单条数据
     * @param id
     */
    public KLine selectByPrimaryKey(Integer id) {
            return this.kLineDao.selectByPrimaryKey(id);
    }
    
    /**
     * 根据条件查询单条记录
     * @param entity 
     */
    public KLine selectByEntity(KLine entity) {
           return this.kLineDao.selectByEntity(entity);
    }
    
    
    /**
     * 根据条件查询多条记录
     * @param entity 
     */
    public List<KLine> selectByEntityList(KLine entity) {
           return  this.kLineDao.selectByEntityList(entity);
    }
    
    /**
     * 根据条件查询Id
     * @param entity 
     */
    public Integer selectById(KLine entity) {
           return  this.kLineDao.selectById(entity);
    }
    
     /**
     * 根据条件查询Ids
     * @param entity 
     */
    public List<Integer> selectByIds(KLine entity) {
        return  this.kLineDao.selectByIds(entity);
    }
    
     /**
     * 根据主键删除单条记录
     * @param id
     */
    public void deleteByPrimaryKey(Integer id) {
           this.kLineDao.deleteByPrimaryKey(id);
    }
    
    /**
     * 根据主键删除多条记录
     * @param list
     */
    public void deleteBatch(List<Integer> list) {
           this.kLineDao.deleteBatch(list);
    }
    
     /**
     * 根据某些条件删除
     * @param entity
     */
    public void deleteBatch(KLine entity) {
           this.kLineDao.deleteByEntity(entity);
    }


    /**
     * 生成日K 8小时 和 12 小时K线数据
     * @param betweenHours
     * @param kLineTypeEnum
     * @param nowDate
     */
    //@Transactional(rollbackFor = Exception.class)
    public void dayLineScheduled(int betweenHours, KLineTypeEnum kLineTypeEnum  ,  LocalDateTime  nowDate) {

        //取到缓存中配置的交易所和交易对:数据结构->"\"{\\\"exchange\\\":
        // {\\\"1\\\":{\\\"symbols\\\":[\\\"BTC/USDT\\\",\\\"BNB/USDT\\\",\\\"ETH/USDT\\\",\\\"QTUM/USDT\\\"]},
        // \\\"2\\\":{\\\"name\\\":\\\"huobipro\\\",\\\"symbols\\\":[\\\"ACT/USDT\\\",\\\"ADA/USDT\\\",\\\"BCH/USDT\\\",\\\"BTM/USDT\\\",\\\"BTS/USDT\\\",\\\"CTXC/USDT\\\",\\\"CVC/USDT\\\",\\\"DASH/USDT\\\",\\\"ELA/USDT\\\",\\\"EOS/USDT\\\",\\\"ETC/USDT\\\",\\\"GNT/USDT\\\",\\\"HSR/USDT\\\",\\\"HT/USDT\\\",\\\"IOST/USDT\\\",\\\"ITC/USDT\\\",\\\"LTC/USDT\\\",\\\"MDS/USDT\\\",\\\"NAS/USDT\\\",\\\"NEO/USDT\\\",\\\"OMG/USDT\\\",\\\"ONT/USDT\\\",\\\"RUFF/USDT\\\",\\\"STORJ/USDT\\\",\\\"THETA/USDT\\\",\\\"TRX/USDT\\\",\\\"VEN/USDT\\\",\\\"XEM/USDT\\\",\\\"XRP/USDT\\\",\\\"ZIL/USDT\\\"]},
        // \\\"13\\\":{\\\"name\\\":{\\\"symbols\\\":[\\\"BTC/USDT\\\",\\\"BNB/USDT\\\",\\\"ETH/USDT\\\",\\\"QTUM/USDT\\\"]},\\\"symbols\\\":[\\\"ZB/USDT\\\"]}}}\""
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
//        LogTrace.info("dayLineScheduled","data",dataString);
        if(StringUtils.isBlank(dataString)){
            return;
        }
        //
        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        if(nowDate == null){
            nowDate = LocalDateTime.now();
        }

        for (String key:exchange.keySet()) {
            try {
                JSONObject jsonObject = exchange.getJSONObject(key);
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    KLine kLine = new KLine();

                    if(nowDate.getHour() == 0){
                        //日K的Date 0点的时候算前一天
                        kLine.setKDate(DateUtils.localDateTimeFormatString(nowDate.minusDays(1),DateUtils.format_ymd));
                    }else{
                        kLine.setKDate(DateUtils.localDateTimeFormatString(nowDate,DateUtils.format_ymd));

                    }
                    kLine.setKType(kLineTypeEnum.getIndex().toString());
                    kLine.setExchangeId(Integer.valueOf(key));

                    kLine.setTransationPairName(symbol);

                    //去除重复数据
                    if(kLineDao.selectByEntity(kLine)!= null){
                        continue;
                    }

                    if(KLineTypeEnum.DAY.getIndex().equals(kLineTypeEnum.getIndex())){
                        // 如果是日K的话，取出12 小时（DAY_HALF1） 和 24小时（DAY_HALF2）的K线计算
                        KLine kLineQuery1 = new KLine();
                        kLineQuery1.setKDate(kLine.getKDate());
                        kLineQuery1.setKType(KLineTypeEnum.DAY_HALF1.getIndex().toString());
                        kLineQuery1.setExchangeId(kLine.getExchangeId());
                        kLineQuery1.setTransationPairName(kLine.getTransationPairName());
                        kLineQuery1 = kLineDao.selectByEntity(kLineQuery1);

                        KLine kLineQuery2 = new KLine();
                        kLineQuery2.setKDate(kLine.getKDate());
                        kLineQuery2.setKType(KLineTypeEnum.DAY_HALF2.getIndex().toString());
                        kLineQuery2.setExchangeId(kLine.getExchangeId());
                        kLineQuery2.setTransationPairName(kLine.getTransationPairName());

                        kLineQuery2 = kLineDao.selectByEntity(kLineQuery2);

                        if(kLineQuery1 == null && kLineQuery2 ==null){
                            continue;
                        }
                        String openPrice ;
                        String closePrice ;
                        String highOpen  ;
                        String lowClose ;
                        String volume;

                        if(kLineQuery1== null && kLineQuery2 != null){
                            openPrice =  kLineQuery2.getOpenPrice();
                            closePrice = kLineQuery2.getClosePrice();
                            highOpen = kLineQuery2.getHighOpen();
                            lowClose = kLineQuery2.getLowClose();
                            volume = kLineQuery2.getVolume();
                        }else if(kLineQuery2== null && kLineQuery1 != null){
                            openPrice =  kLineQuery1.getOpenPrice();
                            closePrice = kLineQuery1.getClosePrice();
                            highOpen = kLineQuery1.getHighOpen();
                            lowClose = kLineQuery1.getLowClose();
                            volume = kLineQuery1.getVolume();
                        }else{
                            openPrice =  kLineQuery1.getOpenPrice();
                            closePrice = kLineQuery2.getClosePrice();
                            highOpen = new BigDecimal(kLineQuery1.getHighOpen()).compareTo(new BigDecimal(kLineQuery2.getHighOpen()))>0 ? kLineQuery1.getHighOpen() : kLineQuery2.getHighOpen();
                            lowClose = new BigDecimal(kLineQuery1.getLowClose()).compareTo(new BigDecimal(kLineQuery2.getLowClose()))<0 ? kLineQuery1.getLowClose() : kLineQuery2.getLowClose();
                            volume = new BigDecimal(kLineQuery1.getVolume()).add(new BigDecimal(kLineQuery2.getVolume())).toString();
                        }
                        kLine.setOpenPrice(openPrice);
                        kLine.setClosePrice(closePrice);
                        kLine.setHighOpen(highOpen);
                        kLine.setLowClose(lowClose);
                        kLine.setVolume(volume);

                        kLine.setExchangeName(jsonObject.getString("name"));
                        kLine.setKName(kLineTypeEnum.getName());
                        kLine.setCreateTime(new Date());
                        kLine.setUpdateTime(new Date());

                        LogTrace.info("K线插入" ,"date1" , kLine.toString());
                        kLineDao.insert(kLine);
                        continue;
                    }


                    kLine.setExchangeName(jsonObject.getString("name"));

                    kLine.setVolume("0");
                    kLine.setKName(kLineTypeEnum.getName());
                    kLine.setCreateTime(new Date());
                    kLine.setUpdateTime(new Date());

                    LocalDateTime date  = nowDate.minusHours(betweenHours);
                    dealKlineData(nowDate, key, symbol, kLine, date);

                    if(StringUtils.isNotBlank(kLine.getOpenPrice())){
                        LogTrace.info("K线插入" ,"date2" , kLine.toString());
                        kLineDao.insert(kLine);
                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
                LogTrace.error(e.getMessage(), e);
            }
        }
    }

//    public static void main(String[] args) {
//        String data = "{\"closePrice\":\"0.15\",\"highestPrice\":\"0.15\",\"lowestPrice\":\"0.149992\",\"openPrice\":\"0.149992\",\"timestamp\":\"1530250337692\",\"volume\":\"0.21\"}";
//        Object listArray = new JSONTokener(data).nextValue();
//        if(listArray instanceof  net.sf.json.JSONArray){
//            JSONArray jsonArray = JSON.parseArray(data);
//            System.out.println("===============================================");
//        }else{
//            JSONObject jsonObject3 = JSON.parseObject(data);
//            TradesToKLine tradesToKLine = jsonObject3.toJavaObject(TradesToKLine.class);
//            System.out.println("----------------------------------------------"+tradesToKLine);
//        }
//    }



    /**
     * 生成小时K线 交易所
     */
    public void klineDataHours() {
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
//        LogTrace.info("dayLineScheduled","data",dataString);
        if(StringUtils.isBlank(dataString)){
            return;
        }

        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        LocalDateTime nowDate = LocalDateTime.now();

        for (String key:exchange.keySet()) {
            LogTrace.infoKvs("一分钟K线数据","key",key);
            try {
                JSONObject jsonObject = exchange.getJSONObject(key);
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    KLine kLine = new KLine();

                    //rowKey规则
                    String rowKey = key + "@" + symbol + "@" + DateUtils.localDateTimeFormatString(nowDate, DateUtils.format_ymdh);

                    kLine.setExchangeId(Integer.valueOf(key));

                    kLine.setTransationPairName(symbol);

                    kLine.setExchangeName(jsonObject.getString("name"));

                    kLine.setVolume("0");
                    kLine.setCreateTime(new Date());
                    kLine.setUpdateTime(new Date());

                    LocalDateTime date  = nowDate.minusHours(1);
                    try {
                        dealKlineData(nowDate, key, symbol, kLine, date);
                    }catch (Exception e){
                        e.printStackTrace();
                        LogTrace.error(e.getMessage(), e);
                    }
                    if(StringUtils.isNotBlank(kLine.getOpenPrice())){
                        LogTrace.info("klineDataHours生成交易所时K" ,"rowKey" , rowKey);
                        currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , kLine);
                        //更改或插入当前时间段的4h和日k的k线数据
                        /*calculateFourHMakeKAction(Integer.valueOf(key),symbol,nowDate,kLine);//时时跟新4h数据
                        calculateByEnumMakeKAction(KLineTypeEnum.DAY,Integer.valueOf(key),symbol,nowDate.minusHours(1),kLine);
                        //如果是日k，更新周k，月k
                        LocalDateTime oldLocal = nowDate.minusHours(1);
                        while (oldLocal.getDayOfWeek() != DayOfWeek.SUNDAY){//本周日时间
                            oldLocal = oldLocal.minusDays(1);
                        }
                        calculateByEnumMakeKAction(KLineTypeEnum.WEEK,Integer.valueOf(key),symbol,oldLocal,kLine);
                        oldLocal = nowDate.minusHours(1);
                        while (oldLocal.getDayOfMonth() != 1){//本月第一天时间
                            oldLocal = oldLocal.minusDays(1);
                        }
                        calculateByEnumMakeKAction(KLineTypeEnum.MONTH,Integer.valueOf(key),symbol,oldLocal,kLine);*/
                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
                LogTrace.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 时时刷新当前k线数据
     * @param kLineTypeEnum
     * @param exchangeId
     * @param transactionName
     * @param time
     * @param kLine
     * @param exchangeName
     */
    private void    calculateByEnumMakeKAction(KLineTypeEnum kLineTypeEnum,Integer exchangeId, String transactionName, LocalDateTime time,KLine kLine){
        try {
            //查询当前交易对日k数据
            //KLineTypeEnum kLineTypeEnum = KLineTypeEnum.DAY_HALF2;
            String exchangeName = kLine.getExchangeName();
            KLine kLine1 = new KLine();
            kLine1.setKDate(DateUtils.localDateTimeFormatString(time,DateUtils.format_ymd));
            kLine1.setKType(kLineTypeEnum.getIndex().toString());
            kLine1.setExchangeId(Integer.valueOf(exchangeId));
            kLine1.setTransationPairName(transactionName);
            kLine1 =kLineDao.selectByEntity(kLine1);
            //如果存在更细，否则直接插入//没有直接插入，否则判断最大、最小、收盘数据，然后更新mysql
            if(kLine1 == null){
                kLine.setKName(kLineTypeEnum.getName());
                kLine.setExchangeName(exchangeName);
                kLine.setKDate(DateUtils.localDateTimeFormatString(time,DateUtils.format_ymd));
                kLine.setKType(kLineTypeEnum.getIndex().toString());
                kLine.setExchangeId(Integer.valueOf(exchangeId));
                kLine.setTransationPairName(transactionName);
                kLine.setCreateTime(new Date());
                kLine.setOpenPrice(new BigDecimal(kLine.getOpenPrice()).stripTrailingZeros().toPlainString());
                kLine.setHighOpen(new BigDecimal(kLine.getHighOpen()).stripTrailingZeros().toPlainString());
                kLine.setLowClose(new BigDecimal(kLine.getLowClose()).stripTrailingZeros().toPlainString());
                kLine.setClosePrice(new BigDecimal(kLine.getClosePrice()).stripTrailingZeros().toPlainString());
                kLine.setVolume(new BigDecimal(kLine.getVolume()).stripTrailingZeros().toPlainString());
                kLineDao.insert(kLine);
            }else{
                BigDecimal volume = new BigDecimal(kLine1.getVolume()).add(new BigDecimal(kLine.getVolume()));
                kLine1.setVolume(volume.stripTrailingZeros().toPlainString());
                //最高价格判断
                BigDecimal bigDecimal = new BigDecimal(kLine.getHighOpen());
                BigDecimal highOpen = new BigDecimal(kLine1.getHighOpen());
                if (bigDecimal.compareTo(highOpen) > 0) {
                    kLine1.setHighOpen(new BigDecimal(kLine.getHighOpen()).stripTrailingZeros().toPlainString());
                }
                //最低价格
                BigDecimal lowCloseNew = new BigDecimal(kLine.getLowClose());
                BigDecimal lowClose = new BigDecimal(kLine1.getLowClose());
                if (lowCloseNew.compareTo(lowClose) < 0) {
                    kLine1.setLowClose(new BigDecimal(kLine.getLowClose()).stripTrailingZeros().toPlainString());
                }
                kLine1.setExchangeName(exchangeName);
                kLine1.setClosePrice(new BigDecimal(kLine.getClosePrice()).stripTrailingZeros().toPlainString());
                kLine1.setUpdateTime(new Date());
                kLine1.setKName(kLineTypeEnum.getName());
                kLineDao.updateByPrimaryKey(kLine1);
            }
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("计算周月k线-时时error",e);
        }
    }
    /**
     * 4h的k线图
     * @param exchangeId
     * @param transactionName
     * @param time
     * @throws Exception
     */
    private void calculateFourHMakeKAction(Integer exchangeId, String transactionName, LocalDateTime time,KLine kLine) throws Exception{
        //得到当前4h时间
        LocalDateTime currencyH = time.minusHours(1);
        while (currencyH.getHour()%4 != 0){
            currencyH = currencyH.minusHours(1);
        }
        //查询当前4h是否已经存在数据
        String rowKeyHead = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_4H;
        String rowKeyOld = rowKeyHead+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(currencyH, DateUtils.format_ymdh);
        TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKeyOld);
        if(tradesToKLine != null){
            BigDecimal volume = new BigDecimal(tradesToKLine.getVolume()).add(new BigDecimal(kLine.getVolume()));
            tradesToKLine.setVolume(volume.stripTrailingZeros().toPlainString());
            tradesToKLine.setClosePrice(kLine.getClosePrice());
            //最高价格
            if (StringUtils.isBlank(tradesToKLine.getHighestPrice())) {
                tradesToKLine.setHighestPrice(kLine.getHighOpen());
            } else {
                BigDecimal bigDecimal = new BigDecimal(kLine.getHighOpen());
                BigDecimal highOpen = new BigDecimal(tradesToKLine.getHighestPrice());
                if (bigDecimal.compareTo(highOpen) > 0) {
                    tradesToKLine.setHighestPrice(kLine.getHighOpen());
                }
            }
            //最低价格
            if (StringUtils.isBlank(tradesToKLine.getLowestPrice())) {
                tradesToKLine.setLowestPrice(kLine.getLowClose());
            } else {
                BigDecimal bigDecimal = new BigDecimal(kLine.getLowClose());
                BigDecimal lowClose = new BigDecimal(tradesToKLine.getLowestPrice());
                if (bigDecimal.compareTo(lowClose) < 0) {
                    tradesToKLine.setLowestPrice(kLine.getLowClose());
                }
            }
        }else{
            while (currencyH.getMinute()%60 !=0){
                currencyH = currencyH.minusMinutes(1);
            }
            tradesToKLine = new TradesToKLine();
            tradesToKLine.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(currencyH).getTime()));//时间错
            tradesToKLine.setOpenPrice(kLine.getOpenPrice());
            tradesToKLine.setHighestPrice(kLine.getHighOpen());
            tradesToKLine.setLowestPrice(kLine.getLowClose());
            tradesToKLine.setClosePrice(kLine.getClosePrice());
            tradesToKLine.setVolume(kLine.getVolume());
        }
        //如果不存在直接插入，否则判断当前数据与老数据的最高最低价格以及收盘价格
        if(StringUtils.isNotBlank(tradesToKLine.getOpenPrice())){
            String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_4H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(currencyH, DateUtils.format_ymdh);
            currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLine);
        }
    }
    /**
     * 处理K线数据
     * @param nowDate 现在时间
     * @param key   交易所ID
     * @param symbol 交易对
     * @param kLine KLine对象
     * @param date  间隔前的时间
     * @throws HbaseConnectionExcetion
     * @throws HbaseGetTableExcetion
     * @throws IOException
     * @throws HbaseCloseExcetion
     */
    private void dealKlineData(LocalDateTime nowDate, String key, String symbol, KLine kLine, LocalDateTime date) throws Exception {
        while (nowDate.compareTo(date)>=0){
            String rowKey = key + "@" + symbol + "@" + DateUtils.localDateTimeFormatString(date, DateUtils.format_ymdhm);

            String data = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                    , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                    , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                    , rowKey);
            if(StringUtils.isNotBlank(data)){
//                try {
//                    Thread.sleep(100);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
                //{"data":"[
                // 1528887840000,  时间戳
                // 6395.69,        开盘
                // 6398.32,        最高价
                // 6390.68,        最低
                // 6396.45,        收盘
                // 9.017711,       交易量
                // 1.40,           涨幅
                // 41571.46,       转人民币价格
                // 6.409           人民币/美元汇率
                // ]","rowKey":"11@BTC/USDT@201806131904"}
                //下标： 0 时间戳 1 开盘 2最高价 3最低  4收盘 5交易量 6涨幅
                LogTrace.infoKvs("一分钟K线数据","rowKey",rowKey,"data",data);
                String openPriceStr ;
                String highOpenStr ;
                String lowCloseStr ;
                String closePriceStr ;
                String volumeStr ;
                Object listArray = new JSONTokener(data).nextValue();
                if(listArray instanceof  net.sf.json.JSONArray){
                    JSONArray jsonArray = JSON.parseArray(data);

                    openPriceStr = jsonArray.getString(1);
                    highOpenStr = jsonArray.getString(2);
                    lowCloseStr  = jsonArray.getString(3);
                    closePriceStr = jsonArray.getString(4);
                    volumeStr = jsonArray.getString(5);

                }else{
                    JSONObject jsonObject3 = JSON.parseObject(data);
                    TradesToKLine tradesToKLine = jsonObject3.toJavaObject(TradesToKLine.class);
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr  = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                }


                BigDecimal volume = new BigDecimal(kLine.getVolume());
                volume = volume.add(new BigDecimal(volumeStr));
                kLine.setVolume(volume.toString());


                if(StringUtils.isBlank(kLine.getOpenPrice() )){
                    kLine.setOpenPrice(new BigDecimal(openPriceStr).stripTrailingZeros().toPlainString());
                }

                kLine.setClosePrice(new BigDecimal(closePriceStr).stripTrailingZeros().toPlainString());

                if(StringUtils.isBlank(kLine.getHighOpen())){
                    kLine.setHighOpen(new BigDecimal(highOpenStr).stripTrailingZeros().toPlainString());
                }else{
                    BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                    BigDecimal highOpen = new BigDecimal(kLine.getHighOpen());
                    if(bigDecimal.compareTo(highOpen)>0){
                        kLine.setHighOpen(new BigDecimal(highOpenStr).stripTrailingZeros().toPlainString());
                    }
                }
                if(StringUtils.isBlank(kLine.getLowClose())){
                    kLine.setLowClose(new BigDecimal(lowCloseStr).stripTrailingZeros().toPlainString());
                }else{
                    BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                    BigDecimal lowClose = new BigDecimal(kLine.getLowClose());
                    if(bigDecimal.compareTo(lowClose)<0 ){
                        kLine.setLowClose(new BigDecimal(lowCloseStr).stripTrailingZeros().toPlainString());
                    }
                }
            }else{
                LogTrace.infoKvs("K线数据为空","rowKey",rowKey,"data",data);
            }
            date = date.plusMinutes(1);
        }
    }

    /**
     * 每周一0点20执行周k数据
     * @param week
     * @param dateDate
     */
    public void weekLineScheduled(KLineTypeEnum week, LocalDateTime dateDate) {
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
        if(StringUtils.isBlank(dataString)){
            return;
        }
        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        if(dateDate == null){
            dateDate = LocalDateTime.now();
        }
        for (String key:exchange.keySet()) {
            try {
                JSONObject jsonObject = exchange.getJSONObject(key);
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                String exchangeName = (String) jsonObject.get("name");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    //分别查询7天前的数据，日k，然后判断最高最低，最后保存数据库中
                    KLine kLine = new KLine();
                    LocalDateTime oldLocal = dateDate.minusDays(7);
                    //循环获取7天日k数据
                    String openPriceStr;
                    String highOpenStr;
                    String lowCloseStr;
                    String closePriceStr;
                    String volumeStr;
                    for(int z=0;z<7;z++){
                        String datastr = DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd);
                        KLine kLineQuery1 = new KLine();
                        kLineQuery1.setKDate(datastr);
                        kLineQuery1.setKType(KLineTypeEnum.DAY.getIndex().toString());
                        kLineQuery1.setExchangeId(Integer.valueOf(key));
                        kLineQuery1.setTransationPairName(symbol);
                        kLineQuery1 = kLineDao.selectByEntity(kLineQuery1);
                        if(kLineQuery1 != null){
                            if(StringUtils.isBlank(kLineQuery1.getOpenPrice())){
                                continue;
                            }
                            if(StringUtils.isBlank(kLineQuery1.getHighOpen())){
                                continue;
                            }
                            if(StringUtils.isBlank(kLineQuery1.getLowClose())){
                                continue;
                            }
                            if(StringUtils.isBlank(kLineQuery1.getClosePrice())){
                                continue;
                            }
                            openPriceStr = kLineQuery1.getOpenPrice();
                            highOpenStr = kLineQuery1.getHighOpen();
                            lowCloseStr = kLineQuery1.getLowClose();
                            closePriceStr = kLineQuery1.getClosePrice();
                            volumeStr = kLineQuery1.getVolume()!=null?kLineQuery1.getVolume():"0";
                            //交易量
                            BigDecimal volume = new BigDecimal(kLine.getVolume()!=null? kLine.getVolume():"0");
                            volume = volume.add(new BigDecimal(volumeStr));
                            kLine.setVolume(volume.toString());
                            //开盘
                            if (StringUtils.isBlank(kLine.getOpenPrice())) {
                                kLine.setOpenPrice(openPriceStr);
                            }
                            //收盘
                            kLine.setClosePrice(closePriceStr);
                            //最高价格
                            if (StringUtils.isBlank(kLine.getHighOpen())) {
                                kLine.setHighOpen(highOpenStr);
                            } else {
                                BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                                BigDecimal highOpen = new BigDecimal(kLine.getHighOpen());
                                if (bigDecimal.compareTo(highOpen) > 0) {
                                    kLine.setHighOpen(highOpenStr);
                                }
                            }
                            //最低价格
                            if (StringUtils.isBlank(kLine.getLowClose())) {
                                kLine.setLowClose(lowCloseStr);
                            } else {
                                BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                                BigDecimal lowClose = new BigDecimal(kLine.getLowClose());
                                if (bigDecimal.compareTo(lowClose) < 0) {
                                    kLine.setLowClose(lowCloseStr);
                                }
                            }
                        }
                        oldLocal = oldLocal.plusDays(1);
                    }
                   /* if("19".equals(key) ){
                        //&& "EOS/BTC".equals(symbol)
                        System.out.println("222");
                    }*/
                    //插入或者更新
                    if(kLine != null && StringUtils.isNotBlank(kLine.getOpenPrice()) && StringUtils.isNotBlank(kLine.getClosePrice())){
                        oldLocal = dateDate.minusDays(7);
                        KLine kLine1 = new KLine();
                        kLine1.setKDate(DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd));
                        kLine1.setKType(week.getIndex().toString());
                        kLine1.setExchangeId(Integer.valueOf(key));
                        kLine1.setTransationPairName(symbol);
                        kLine1 =kLineDao.selectByEntity(kLine1);
                        //如果存在更细，否则直接插入
                        if(kLine1 == null){
                            kLine.setKName(week.getName());
                            kLine.setExchangeName(exchangeName);
                            kLine.setKDate(DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd));
                            kLine.setKType(week.getIndex().toString());
                            kLine.setExchangeId(Integer.valueOf(key));
                            kLine.setTransationPairName(symbol);
                            kLine.setCreateTime(new Date());
                            kLineDao.insert(kLine);
                        }else{
                            kLine1.setExchangeName(exchangeName);
                            kLine1.setOpenPrice(kLine.getOpenPrice());
                            kLine1.setHighOpen(kLine.getHighOpen());
                            kLine1.setLowClose(kLine.getLowClose());
                            kLine1.setClosePrice(kLine.getClosePrice());
                            kLine1.setUpdateTime(new Date());
                            kLine1.setKName(week.getName());
                            kLineDao.updateByPrimaryKey(kLine1);
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
                LogTrace.error(e.getMessage(), e);
            }
        }
    }

    public void monthLineScheduled(KLineTypeEnum month, LocalDateTime dateDate) {
        String dataString = redisServer.get(RedisKeysUtils.EXCHANGE_SYMBOLS);
        if(StringUtils.isBlank(dataString)){
            return;
        }
        JSONObject exchange = JSON.parseObject(dataString).getJSONObject(RedisKeysUtils.EXCHANGE);
        if(dateDate == null){
            dateDate = LocalDateTime.now();
        }
        LocalDateTime localDateTime = dateDate.minusDays(1);
        int mouthDay = DateUtils.getDayOfMonth(localDateTime.getYear(),localDateTime.getMonthValue());//本月总天数
        for (String key:exchange.keySet()) {
            try {
                JSONObject jsonObject = exchange.getJSONObject(key);
                JSONArray symbols = jsonObject.getJSONArray("symbols");
                String exchangeName = (String) jsonObject.get("name");
                for (int i = 0; i < symbols.size(); i++) {
                    String symbol = symbols.getString(i);
                    //分别查询7天前的数据，日k，然后判断最高最低，最后保存数据库中
                    KLine kLine = new KLine();
                    LocalDateTime oldLocal = dateDate.minusDays(mouthDay);
                    //循环获取前一月k数据
                    String openPriceStr;
                    String highOpenStr;
                    String lowCloseStr;
                    String closePriceStr;
                    String volumeStr;
                    for(int z=0;z<mouthDay;z++){
                        String datastr = DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd);
                        KLine kLineQuery1 = new KLine();
                        kLineQuery1.setKDate(datastr);
                        kLineQuery1.setKType(KLineTypeEnum.DAY.getIndex().toString());
                        kLineQuery1.setExchangeId(Integer.valueOf(key));
                        kLineQuery1.setTransationPairName(symbol);
                        kLineQuery1 = kLineDao.selectByEntity(kLineQuery1);
                        if(kLineQuery1 != null){
                            if(StringUtils.isBlank(kLineQuery1.getOpenPrice())){
                                continue;
                            }
                            if(StringUtils.isBlank(kLineQuery1.getHighOpen())){
                                continue;
                            }
                            if(StringUtils.isBlank(kLineQuery1.getLowClose())){
                                continue;
                            }
                            if(StringUtils.isBlank(kLineQuery1.getClosePrice())){
                                continue;
                            }
                            openPriceStr = kLineQuery1.getOpenPrice();
                            highOpenStr = kLineQuery1.getHighOpen();
                            lowCloseStr = kLineQuery1.getLowClose();
                            closePriceStr = kLineQuery1.getClosePrice();
                            volumeStr = kLineQuery1.getVolume()!=null?kLineQuery1.getVolume():"0";
                            //交易量
                            BigDecimal volume = new BigDecimal(kLine.getVolume()!=null? kLine.getVolume():"0");
                            volume = volume.add(new BigDecimal(volumeStr));
                            kLine.setVolume(volume.toString());
                            //开盘
                            if (StringUtils.isBlank(kLine.getOpenPrice())) {
                                kLine.setOpenPrice(openPriceStr);
                            }
                            //收盘
                            kLine.setClosePrice(closePriceStr);
                            //最高价格
                            if (StringUtils.isBlank(kLine.getHighOpen())) {
                                kLine.setHighOpen(highOpenStr);
                            } else {
                                BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                                BigDecimal highOpen = new BigDecimal(kLine.getHighOpen());
                                if (bigDecimal.compareTo(highOpen) > 0) {
                                    kLine.setHighOpen(highOpenStr);
                                }
                            }
                            //最低价格
                            if (StringUtils.isBlank(kLine.getLowClose())) {
                                kLine.setLowClose(lowCloseStr);
                            } else {
                                BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                                BigDecimal lowClose = new BigDecimal(kLine.getLowClose());
                                if (bigDecimal.compareTo(lowClose) < 0) {
                                    kLine.setLowClose(lowCloseStr);
                                }
                            }
                        }
                        oldLocal = oldLocal.plusDays(1);
                    }
                    if("19".equals(key) ){
                        //&& "EOS/BTC".equals(symbol)
                        System.out.println("222");
                    }
                    //插入或者更新
                    if(kLine != null && StringUtils.isNotBlank(kLine.getOpenPrice()) && StringUtils.isNotBlank(kLine.getClosePrice())){
                        oldLocal = dateDate.minusDays(mouthDay);
                        KLine kLine1 = new KLine();
                        kLine1.setKDate(DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd));
                        kLine1.setKType(month.getIndex().toString());
                        kLine1.setExchangeId(Integer.valueOf(key));
                        kLine1.setTransationPairName(symbol);
                        kLine1 =kLineDao.selectByEntity(kLine1);
                        //如果存在更细，否则直接插入
                        if(kLine1 == null){
                            kLine.setKName(month.getName());
                            kLine.setExchangeName(exchangeName);
                            kLine.setKDate(DateUtils.localDateTimeFormatString(oldLocal,DateUtils.format_ymd));
                            kLine.setKType(month.getIndex().toString());
                            kLine.setExchangeId(Integer.valueOf(key));
                            kLine.setTransationPairName(symbol);
                            kLine.setCreateTime(new Date());
                            kLineDao.insert(kLine);
                        }else{
                            kLine1.setExchangeName(exchangeName);
                            kLine1.setOpenPrice(kLine.getOpenPrice());
                            kLine1.setHighOpen(kLine.getHighOpen());
                            kLine1.setLowClose(kLine.getLowClose());
                            kLine1.setClosePrice(kLine.getClosePrice());
                            kLine1.setUpdateTime(new Date());
                            kLine1.setKName(month.getName());
                            kLineDao.updateByPrimaryKey(kLine1);
                        }
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
                LogTrace.error(e.getMessage(), e);
            }
        }
    }

    /**
     * 获取hbase数据（根据rowkey获取1hk线数据）
     * @param rowKey
     * @return
     * @throws Exception
     */
    public TradesToKLine getHbaseDataByRowKey(String rowKey) throws Exception{
        Map<String, Object> hbaseData = hbaseBaseQueryComponent.selectData(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME
                , rowKey);
        if(null != hbaseData && hbaseData.get("volume") != null && hbaseData.get("openPrice") != null && hbaseData.get("highestPrice") != null&& hbaseData.get("lowestPrice") != null&& hbaseData.get("closePrice") != null){
            TradesToKLine tradesToKLine = new TradesToKLine();
            tradesToKLine.setTimestamp(hbaseData.get("timestamp").toString());
            tradesToKLine.setVolume(hbaseData.get("volume").toString());
            tradesToKLine.setOpenPrice(hbaseData.get("openPrice").toString());
            tradesToKLine.setClosePrice(hbaseData.get("closePrice").toString());
            tradesToKLine.setHighestPrice(hbaseData.get("highestPrice").toString());
            tradesToKLine.setLowestPrice(hbaseData.get("lowestPrice").toString());
            return tradesToKLine;
        }
        return null;
    }
}